import math
import traceback
from datetime import datetime, timedelta

import numpy as np
import pandas as pd
from dateutil.relativedelta import relativedelta

from air_web.config.config import config
from air_web.dw.data_mapping import ConsType, DataType
from air_web.web_flask.bll.base_bll import BaseBll
from air_web.web_flask.bll.model_map import ModelMap
from air_web.web_flask.common.constants import ReturnType


class UserPredictBll(BaseBll):
    """总览页面视图函数业务逻辑类"""

    time_field = "data_time"
    cons_no_field = "cons_no"
    C_CONS = config.get("C_CONS", "c_cons")

    def __init__(self):
        """
        如果当前页面用到mysql则init_mysql_dal=True，self.sql_dal就会提供mysql实例，不用到则传False
        如果当前页面用到es则init_es_dal=True，self.es_dal就会提供es实例，不用到则传False
        """
        super().__init__(
            init_mysql_dal=True, init_es_dal=True, init_cons_mysql_dal=True
        )
        self.today = datetime.now().strftime("%Y-%m-%d")
        self.year = str(datetime.now().year)

    def get_cons_num(self, cons_no=None, cons_name=None):
        rules = []
        where_sql_list = []
        where_sql = ""

        if cons_no is not None:
            where_sql_list.append(f" cons_no like '%%{cons_no}%%' ")

        if cons_name is not None:
            where_sql_list.append(f" cons_name like '%%{cons_name}%%' ")

        if len(where_sql_list) >= 1:
            where_sql = " where " + " and ".join(where_sql_list)

        table_name = self.C_CONS
        sql = (
            f"""select {self.cons_no_field}, on5,on7 from {table_name} """.format(
                table_name=table_name
            )
            + where_sql
        )
        res_df = self.sql_dal.sql_pandas.query(sql)
        if not len(res_df):
            self.logger.warning(f"查询结果为空:{table_name}, {rules}")
            return res_df
        self.logger.info(f"查询:{table_name},结果数量:{len(res_df)}, {rules}")
        res_df = res_df.to_dict("records")
        return res_df

    def get_table_name(self, cons_no, on7, Flag=None):
        if Flag == False:
            if self.data_source == "es":
                table_name = config[
                    "POWERLOAD_PREDICT_PRE_IDX"
                ] + "*-{}".format(on7)
            else:
                table_name = config.get(
                    "ACLR_RES_PREDICT_96", "aclr_res_predict_96"
                )
        else:
            if self.data_source == "es":
                powerload_idx = config["POWERLOAD"]
                on7 = str(ModelMap.get("c_cons_all", "on7", db_key=cons_no))
                is_cal = "1" if config["c_cons_filter_is_cal"] else "*"
                table_name = (
                    powerload_idx.replace("on5", "*")
                    .replace("on7", on7)
                    .replace("cal01", is_cal)
                )
            else:
                table_name = config.get(
                    "ACLR_RES_POWER_96", "ACLR_RES_POWER_96"
                )
        self.logger.info("要查询的powerload索引:{}".format(table_name))
        return table_name

    def get_predict_data(
        self, cons_no, on5, on7, start_date=None, end_date=None, Flag=None
    ):
        try:
            rules = []
            table_name = self.get_table_name(
                cons_no=cons_no, on7=on7, Flag=Flag
            )

            if self.data_source == "es":
                doc_time_field = "data_time"
            else:
                doc_time_field = "data_date"
                rules.append(
                    (
                        "data_type",
                        "query",
                        "in",
                        [DataType.P_TOTAL, DataType.P_KT],
                    )
                )

            if cons_no is not None:
                rules.append((self.cons_no_field, "query", "=", cons_no))
            if start_date is not None and end_date is not None:
                rules.extend(
                    [
                        (doc_time_field, "query", ">=", str(start_date)),
                        (doc_time_field, "query", "<", str(end_date)),
                    ]
                )

            df_result = self.ana_query_all(
                start_date,
                end_date,
                table_name,
                rules,
                self.data_source,
                doc_time_field,
                True,
                True,
            )
            if not len(df_result):
                return pd.DataFrame(columns=["data_time", "p_total", "p_kt"])
            df_result.sort_values(
                by=["data_time"], ascending=True, inplace=True
            )
            df_result.rename(
                columns={"data_time": "compute_time"}, inplace=True
            )
            df_result["kt_pct"] = round(
                df_result["p_kt"] / df_result["p_total"], 4
            )
            df_result["kt_pct"] = df_result["kt_pct"].fillna(0)
            df_result = df_result.astype(str)
            df_result.drop_duplicates(
                subset="compute_time", inplace=True, keep="last"
            )

            return df_result
        except:
            self.logger.error(traceback.format_exc())
            return pd.DataFrame()

    def get_download_data(self, res_df, cons_name, start_date):
        if len(res_df) <= 0:
            res_df["compute_time"] = [
                (
                    datetime.strptime("00:00", "%H:%M")
                    + relativedelta(minutes=15 * i)
                ).strftime("%H:%M")
                for i in range(96)
            ]
        res_df = res_df.set_index("compute_time").transpose()
        res_df["cons_name"] = cons_name
        res_df["date"] = start_date
        res_df["type"] = res_df.index.map(
            lambda x: "总负荷" if "total" in x else "空调负荷" if "kt" in x else None
        )
        res_df["data_type"] = res_df.index.map(
            lambda x: "实际值" if "_true" in x else "预测值" if "_pre" in x else None
        )
        res_df = res_df.reset_index(drop=True)
        res_df.sort_values(
            ["type", "data_type"], ascending=(True, False), inplace=True
        )
        res_df.dropna(subset=["type"], inplace=True)

        return res_df

    def get_all_result(self, params, is_download):
        cons_no = params.get("cons_no")
        cons_name = params.get("cons_name")
        start_str = params.get("start_date")
        df = self.get_cons_num(cons_no=cons_no)
        on5 = df[0].get("on5") if len(df) >= 1 else ""
        on7 = df[0].get("on7") if len(df) >= 1 else ""
        cons_no = df[0].get(self.cons_no_field) if len(df) >= 1 else ""
        start_date = datetime.strptime(start_str, "%Y-%m-%d")

        # 如果展示数据是今天，那么end_time取现在(因为用户计算结果中存了全天的预测点)
        if start_date.date() == datetime.now().date():
            end_date = datetime.now()
        else:
            end_date = start_date + timedelta(days=1)

        predict_data = self.get_predict_data(
            cons_no=cons_no,
            on5=on5,
            on7=on7,
            start_date=start_date,
            end_date=end_date,
            Flag=False,
        )
        real_data = self.get_predict_data(
            cons_no=cons_no,
            on5=on5,
            on7=on7,
            start_date=start_date,
            end_date=end_date,
            Flag=True,
        )
        if len(real_data) >= 1 and len(predict_data) >= 1:
            res_df = pd.merge(
                real_data,
                predict_data,
                on="compute_time",
                suffixes=("_true", "_pre"),
                how="outer",
            )
            res_df = res_df.round(2)
            res_df = res_df.where(res_df.notnull(), None)
            res_df["compute_time"] = res_df["compute_time"].apply(
                lambda x: x[-8:-3]
            )
            res_df = res_df[
                [
                    "compute_time",
                    "p_total_pre",
                    "p_kt_pre",
                    "p_total_true",
                    "p_kt_true",
                ]
            ]
        elif len(real_data) >= 1 and len(predict_data) == 0:
            res_df = real_data
            res_df = res_df.round(2)
            res_df = res_df.where(res_df.notnull(), None)
            res_df = res_df[["compute_time", "p_total", "p_kt"]]
            res_df["compute_time"] = res_df["compute_time"].apply(
                lambda x: x[-8:-3]
            )
            res_df.rename(
                columns={"p_total": "p_total_true", "p_kt": "p_kt_true"},
                inplace=True,
            )
            res_df["p_total_pre"] = None
            res_df["p_kt_pre"] = None
        elif len(real_data) == 0 and len(predict_data) >= 1:
            res_df = predict_data
            res_df = res_df.round(2)
            res_df = res_df.where(res_df.notnull(), None)
            res_df = res_df[["compute_time", "p_total", "p_kt"]]
            res_df["compute_time"] = res_df["compute_time"].apply(
                lambda x: x[-8:-3]
            )
            res_df.rename(
                columns={"p_total": "p_total_pre", "p_kt": "p_kt_pre"},
                inplace=True,
            )
            res_df["p_total_true"] = None
            res_df["p_kt_true"] = None
        else:
            res_df = pd.DataFrame(
                columns=[
                    "data_time",
                    "p_total_pre",
                    "p_kt_pre",
                    "p_total_true",
                    "p_kt_true",
                ]
            )

        res_df["p_total_true"] = res_df["p_total_true"].astype(float)
        res_df["p_total_pre"] = res_df["p_total_pre"].astype(float)
        res_df["p_kt_true"] = res_df["p_kt_true"].astype(float)
        res_df["p_kt_pre"] = res_df["p_kt_pre"].astype(float)
        res_df = res_df.round(2)
        res_df = res_df.replace({np.nan: None})
        return_dict = res_df.to_dict("records")
        if is_download:
            return (
                self.get_download_data(res_df, cons_name, start_date),
                None,
            )  # DataFrame
        total_rate, kt_rate, _ = self.get_predict_rate(res_df)
        statistic_data = {"total_rate": total_rate, "kt_rate": kt_rate}
        return return_dict, statistic_data

    def get_user_cons(self, param):
        cons_name = param.get("cons_name")
        try:
            where_sql = ""
            where_sql_list = []
            if cons_name:
                where_sql_list.append(
                    f" cons_name like '%%{cons_name}%%' or cons_no like '%%{cons_name}%%' "
                )

            if len(where_sql_list) >= 1:
                where_sql = " and " + " and ".join(where_sql_list)
            sql = """select cons_no as real_cons_no, cons_name
                     from {c_cons} where is_remove = 0 and cons_type={cons_type}
                     {where_sql}
                     order by data_rank asc limit 100
                  """.format(
                c_cons=self.C_CONS,
                cons_name=cons_name,
                where_sql=where_sql,
                cons_type=ConsType.ORDINARY_USER,
            )
            res_df = self.sql_dal.sql_pandas.query(sql)
            res_dict = res_df.to_dict("records")
            return res_dict
        except:
            self.logger.error(traceback.format_exc())
            return ReturnType.ERROR
